package org.chnxi.study.messagequeue.kafka.comsumer;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class SendedToMsgDetailPollMsgProcessor implements PollMsgProcessor<String,String> {

    @Autowired
    private ConsumeMsgProcessor<String, String> consumeMsgProcessor;

    @Autowired
    private OffsetProcessor<String, String> offsetProcessor;

    public void pollMsg(KafkaConsumer<String,String> consumer){

        while (true){
            // 从kafka中拉取数据
            ConsumerRecords<String, String> consumerRecords = consumer.poll(300L);

            // 未能成功拉取到消息，尝试下次拉取
            if(null == consumerRecords || consumerRecords.isEmpty()){
                continue;
            }

            // 获取本次拉取消息相关的offset
            Map<TopicPartition, Long> offset = new HashMap();
            consumerRecords.forEach(record -> {
                String topic = record.topic();
                int partition = record.partition();
                long newOffset = record.offset();
                TopicPartition topicPartition = new TopicPartition(topic, partition);
                offset.put(topicPartition, newOffset);
            });

            // 将消息数据交给msgProcessor处理
            boolean consume = consumeMsgProcessor.consume(consumerRecords);

            if(consume){
                consumer.commitSync();
                offsetProcessor.saveOffSet(offset);
            }
        }
    }

}
